package com.nd.te.test;

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.util.Objects;

public class datahar {

    public static class mapper extends Mapper<LongWritable, Text, globalterror, IntWritable>  {
        protected void map(LongWritable index, Text line, Context context) throws IOException,InterruptedException {
            String[] cc = line.toString().split(",");
            globalterror gt =new globalterror();
            if(cc.length == 9){
                if(cc[8]!="1"&&cc[8]!="2"&cc[8]!="3"&&cc[8]!="4"&&cc[8]!="5"&&cc[8]!="6"&&cc[8]!="7"&cc[8]!="8"&&cc[8]!="9") {
                    cc[8]="10";
                }
                gt.set(cc[2],Integer.parseInt(cc[0]),Integer.parseInt(cc[8]));
                context.write(gt,new IntWritable(1));
            }
        }
    }

    public static class reducer extends Reducer<globalterror,IntWritable,globalterror,IntWritable>{
        protected void reduce(globalterror key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException{
            int count =0;
            for (IntWritable e: values) {
                count += e.get();
            }
            context.write(key, new IntWritable(count));
        }
    }

    public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {
        //args=new String[]{"hdfs://hadoop101:8020/wc/input/data.txt","hdfs://hadoop101:8020/wc/datapro1.txt"};
        //args = new String[]{"C:\\Users\\张传畅\\Desktop\\bigwork\\data.txt", "C:\\Users\\张传畅\\Desktop\\bigwork\\out"};
        args=new String[]{"data/input/dataprocess.txt","data/dataselect1"};

        Configuration configuration = new Configuration();
        Job job = Job.getInstance(configuration);

        job.setMapperClass(mapper.class);
        //job.setGroupingComparatorClass(grouper.class);
        job.setReducerClass(reducer.class);

        job.setMapOutputKeyClass(globalterror.class);
        job.setMapOutputValueClass(IntWritable.class);

        job.setOutputKeyClass(globalterror.class);
        job.setOutputValueClass(IntWritable.class);

        FileInputFormat.addInputPath(job,new Path(args[0]));
        FileOutputFormat.setOutputPath(job,new Path(args[1]));

        boolean result = job.waitForCompletion(true);
        System.exit(result ? 0 : 1);
    }
}
